// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: authentication-free shared key and token derivation between clients
// feature: DServer support simple entity
// feature: provide query interfaces
// feature: develop dispatcher
// feature: Support DServer-DClient communication
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: discovery support client and server
// feature: evbs support report hostname processname and version
// feature: dds support get remote participant vbs version and rm some print
// feature: add startup time in PDP msg
// ------------------------------------------------------------------

#include <edds/rtps/builtin/data/ParticipantProxyData.h>

#include <edds/rtps/builtin/BuiltinProtocols.h>
#include <edds/rtps/builtin/data/WriterProxyData.h>
#include <edds/rtps/builtin/data/ReaderProxyData.h>
#include <discovery/participant/PDPSimple.h>
#include <transport/TransportFactory.hpp>
#include <edds/rtps/resources/TimedEvent.h>
#include <ertps/utils/TimeConversion.h>

#include <rtps/participant/RTPSParticipantImpl.h>
#include <xmlparser/XMLParserCommonPart.h>

#include <mutex>
#include <chrono>

#include <elog/Log.hpp>
#include <edds/dds/core/policy/ParameterList.hpp>
#include <edds/dds/core/policy/QosPoliciesSerializer.hpp>

#include "ProxyDataFilters.hpp"
#include "ProxyHashTables.hpp"

using namespace evbs::ertps;
using ParameterList = evbs::edds::dds::ParameterList;

namespace evbs {
namespace ertps {
namespace rtps {

ParticipantProxyData::ParticipantProxyData(const RTPSParticipantAllocationAttributes& allocation)
    : m_protocolVersion(c_ProtocolVersion),
      m_VendorId(c_VendorId_Unknown),
      m_domain_id(edds::dds::DOMAIN_ID_UNKNOWN),
      m_availableBuiltinEndpoints(0U),
      metatraffic_locators(allocation.locators.max_unicast_locators, allocation.locators.max_multicast_locators),
      default_locators(allocation.locators.max_unicast_locators, allocation.locators.max_multicast_locators),
      m_manualLivelinessCount()
#if HAVE_SECURITY
      ,
      security_attributes_(0UL),
      plugin_security_attributes_(0UL)
#endif  // if HAVE_SECURITY
      ,
      isAlive(false),
      m_properties(static_cast<uint32_t>(allocation.data_limits.max_properties)),
      lease_duration_event(nullptr),
      should_check_lease_duration(false),
      m_readers(new ProxyHashTable<ReaderProxyData>(allocation.readers)),
      m_writers(new ProxyHashTable<WriterProxyData>(allocation.writers)),
      m_sample_identity(),
      transport_type(TransportTypeKind::TRANSPORT_TYPE_UDP),
      lease_duration_(std::chrono::microseconds::zero()) {
}

ParticipantProxyData::ParticipantProxyData(const ParticipantProxyData& pdata)
    : m_protocolVersion(pdata.m_protocolVersion),
      m_guid(pdata.m_guid),
      m_VendorId(pdata.m_VendorId),
      m_domain_id(pdata.m_domain_id),
      m_availableBuiltinEndpoints(pdata.m_availableBuiltinEndpoints),
      metatraffic_locators(pdata.metatraffic_locators),
      default_locators(pdata.default_locators),
      m_manualLivelinessCount(),
      m_participantName(pdata.m_participantName),
      m_key(pdata.m_key),
      m_leaseDuration(pdata.m_leaseDuration)
#if HAVE_SECURITY
      ,
      identity_token_(pdata.identity_token_),
      permissions_token_(pdata.permissions_token_),
      security_attributes_(pdata.security_attributes_),
      plugin_security_attributes_(pdata.plugin_security_attributes_),
      m_auth_token(pdata.m_auth_token),
      is_client(pdata.is_client)
#endif  // if HAVE_SECURITY
      ,
      isAlive(pdata.isAlive),
      m_properties(pdata.m_properties),
      lease_duration_event(nullptr),
      should_check_lease_duration(false)
      // This method is only called when calling the participant discovery listener and the
      // corresponding DiscoveredParticipantInfo struct is created. Only participant info is used,
      // so there is no need to copy m_readers and m_writers
      ,
      m_readers(nullptr),
      m_writers(nullptr),
      m_sample_identity(pdata.m_sample_identity),
      m_startTime(pdata.m_startTime),
      m_evbsRemoteVersion(pdata.m_evbsRemoteVersion),
      m_hostName(pdata.m_hostName),
      transport_type(pdata.transport_type),
      isCompatible(pdata.isCompatible),
      lease_duration_(pdata.lease_duration_),
      m_hostId(pdata.m_hostId),
      m_pid(pdata.m_pid) {
}

ParticipantProxyData::~ParticipantProxyData() {
    logDebug(RTPS_PARTICIPANT, "ParticipantProxy destructor, guid: " << m_guid << " name: " << m_participantName);

    // delete all reader proxies
    if (m_readers) {
        for (ProxyHashTable<ReaderProxyData>::value_type val : *m_readers) {
            delete val.second;
        }

        delete m_readers;
    }

    // delete all writers proxies
    if (m_writers) {
        for (ProxyHashTable<WriterProxyData>::value_type val : *m_writers) {
            delete val.second;
        }

        delete m_writers;
    }

    if (lease_duration_event != nullptr) {
        delete lease_duration_event;
    }
}

uint32_t ParticipantProxyData::get_serialized_size(bool include_encapsulation) const {
    uint32_t ret_val = include_encapsulation ? 4U : 0U;

    // PID_PROTOCOL_VERSION
    ret_val += 4U + 4U;

    // PID_VENDORID
    ret_val += 4U + 4U;

    // PID_DOMAIN_ID
    ret_val += 4U + 4U;

    // PID_PARTICIPANT_GUID
    ret_val += 4U + PARAMETER_GUID_LENGTH;

    // PID_METATRAFFIC_MULTICAST_LOCATOR
    ret_val += static_cast<uint32_t>((4U + PARAMETER_LOCATOR_LENGTH) * metatraffic_locators.multicast.size());

    // PID_METATRAFFIC_UNICAST_LOCATOR
    ret_val += static_cast<uint32_t>((4U + PARAMETER_LOCATOR_LENGTH) * metatraffic_locators.unicast.size());

    // PID_DEFAULT_UNICAST_LOCATOR
    ret_val += static_cast<uint32_t>((4U + PARAMETER_LOCATOR_LENGTH) * default_locators.unicast.size());

    // PID_DEFAULT_MULTICAST_LOCATOR
    ret_val += static_cast<uint32_t>((4U + PARAMETER_LOCATOR_LENGTH) * default_locators.multicast.size());

    // PID_PARTICIPANT_LEASE_DURATION
    ret_val += 4U + PARAMETER_TIME_LENGTH;

    // PID_BUILTIN_ENDPOINT_SET
    ret_val += 4U + PARAMETER_BUILTINENDPOINTSET_LENGTH;

    if (m_participantName.size() > 0U) {
        // PID_ENTITY_NAME
        ret_val += edds::dds::ParameterSerializer<Parameter_t>::cdr_serialized_size(m_participantName);
    }

    if (m_properties.size() > 0U) {
        // PID_PROPERTY_LIST
        ret_val += edds::dds::ParameterSerializer<ParameterPropertyList_t>::cdr_serialized_size(m_properties);
    }

#if HAVE_SECURITY
    if (!identity_token_.class_id().empty()) {
        // PID_IDENTITY_TOKEN
        ret_val += edds::dds::ParameterSerializer<Parameter_t>::cdr_serialized_size(identity_token_);
    }

    if (!permissions_token_.class_id().empty()) {
        // PID_PERMISSIONS_TOKEN
        ret_val += edds::dds::ParameterSerializer<Parameter_t>::cdr_serialized_size(permissions_token_);
    }

    if ((security_attributes_ != 0UL) || (plugin_security_attributes_ != 0UL)) {
        // PID_PARTICIPANT_SECURITY_INFO
        ret_val += 4 + PARAMETER_PARTICIPANT_SECURITY_INFO_LENGTH;
    }
#endif  // if HAVE_SECURITY

    // PID_STARTUP
    ret_val += 4U + PARAMETER_TIME_LENGTH;

    // PID_EVBS_VERSION
    ret_val += 4 + PARAMETER_EVBS_VERSION_LENGTH;

    //PID_TRANSPORT_TYPE
    ret_val += 4U + 4U;

    if (m_hostName.size() > 0U) {
        // PID_HOST_NAME
        ret_val += edds::dds::ParameterSerializer<Parameter_t>::cdr_serialized_size(m_hostName);
    }

    if (isCompatible) {
        // PID_COMPATIBLE
        ret_val += 4U + PARAMETER_BOOL_LENGTH;
    }
    if (m_guid.guidPrefix.is_guid_static()) {
        // PID_HOSTID_PID_INFO
        ret_val += 4U + PARAMETER_HOSTID_PID_INFO_LENGTH;
    }

    // PID_SENTINEL
    return ret_val + 4U;
}

bool ParticipantProxyData::writeToCDRMessage(CDRMessage_t* msg, bool write_encapsulation) {
    if (write_encapsulation) {
        if (!ParameterList::writeEncapsulationToCDRMsg(msg)) {
            return false;
        }
    }

    {
        ParameterProtocolVersion_t p(edds::dds::PID_PROTOCOL_VERSION, 4U);
        p.protocolVersion = this->m_protocolVersion;
        if (!edds::dds::ParameterSerializer<ParameterProtocolVersion_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }
    {
        ParameterVendorId_t p(edds::dds::PID_VENDORID, 4U);
        p.vendorId[0] = this->m_VendorId[0];
        p.vendorId[1] = this->m_VendorId[1];
        if (!edds::dds::ParameterSerializer<ParameterVendorId_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }
    {
        ParameterDomainId_t p(edds::dds::PID_DOMAIN_ID, 4);
        p.domain_id = this->m_domain_id;
        if (!edds::dds::ParameterSerializer<ParameterDomainId_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }
    {
        ParameterGuid_t p(edds::dds::PID_PARTICIPANT_GUID, PARAMETER_GUID_LENGTH, m_guid);
        if (!edds::dds::ParameterSerializer<ParameterGuid_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }
    for (const Locator_t& it : metatraffic_locators.multicast) {
        ParameterLocator_t p(edds::dds::PID_METATRAFFIC_MULTICAST_LOCATOR, PARAMETER_LOCATOR_LENGTH, it);
        if (!edds::dds::ParameterSerializer<ParameterLocator_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }
    for (const Locator_t& it : metatraffic_locators.unicast) {
        ParameterLocator_t p(edds::dds::PID_METATRAFFIC_UNICAST_LOCATOR, PARAMETER_LOCATOR_LENGTH, it);
        if (!edds::dds::ParameterSerializer<ParameterLocator_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }
    for (const Locator_t& it : default_locators.unicast) {
        ParameterLocator_t p(edds::dds::PID_DEFAULT_UNICAST_LOCATOR, PARAMETER_LOCATOR_LENGTH, it);
        if (!edds::dds::ParameterSerializer<ParameterLocator_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }
    for (const Locator_t& it : default_locators.multicast) {
        ParameterLocator_t p(edds::dds::PID_DEFAULT_MULTICAST_LOCATOR, PARAMETER_LOCATOR_LENGTH, it);
        if (!edds::dds::ParameterSerializer<ParameterLocator_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }
    {
        ParameterTime_t p(edds::dds::PID_PARTICIPANT_LEASE_DURATION, PARAMETER_TIME_LENGTH);
        p.time = m_leaseDuration;
        if (!edds::dds::ParameterSerializer<ParameterTime_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }
    {
        ParameterBuiltinEndpointSet_t p(edds::dds::PID_BUILTIN_ENDPOINT_SET, PARAMETER_BUILTINENDPOINTSET_LENGTH);
        p.endpointSet = m_availableBuiltinEndpoints;
        if (!edds::dds::ParameterSerializer<ParameterBuiltinEndpointSet_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }

    if (m_participantName.size() > 0U) {
        ParameterString_t p(edds::dds::PID_ENTITY_NAME, 0, m_participantName);
        if (!edds::dds::ParameterSerializer<ParameterString_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }

    if (m_properties.size() > 0U) {
        if (!edds::dds::ParameterSerializer<ParameterPropertyList_t>::add_to_cdr_message(m_properties, msg)) {
            return false;
        }
    }

#if HAVE_SECURITY
    if (!identity_token_.class_id().empty()) {
        ParameterToken_t p(edds::dds::PID_IDENTITY_TOKEN, 0U);
        p.token = identity_token_;
        if (!edds::dds::ParameterSerializer<ParameterToken_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }

    if (!permissions_token_.class_id().empty()) {
        ParameterToken_t p(edds::dds::PID_PERMISSIONS_TOKEN, 0U);
        p.token = permissions_token_;
        if (!edds::dds::ParameterSerializer<ParameterToken_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }

    if ((security_attributes_ != 0UL) || (plugin_security_attributes_ != 0UL)) {
        ParameterParticipantSecurityInfo_t p;
        p.security_attributes = security_attributes_;
        p.plugin_security_attributes = plugin_security_attributes_;
        if (!edds::dds::ParameterSerializer<ParameterParticipantSecurityInfo_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }
#endif  // if HAVE_SECURITY

    {
        ParameterTime_t p(edds::dds::PID_STARTUP, PARAMETER_TIME_LENGTH);
        p.time = m_startTime;
        if (!edds::dds::ParameterSerializer<ParameterTime_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }

    {
        ParameterVersion_t p(edds::dds::PID_EVBS_VERSION, 4U);
        p.evbsVersion = edds::rtps::c_Version_vbs;
        if (!edds::dds::ParameterSerializer<ParameterVersion_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }

    {
        ParameterTransportType_t p;
        p.type = transport_type;
        if (!edds::dds::ParameterSerializer<ParameterTransportType_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }

    if (m_hostName.size() > 0U) {
        ParameterString_t p(edds::dds::PID_HOST_NAME, 0, m_hostName);
        if (!edds::dds::ParameterSerializer<ParameterString_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }

    if (isCompatible) {
        ParameterBool_t p(edds::dds::PID_COMPATIBLE, PARAMETER_BOOL_LENGTH, isCompatible);
        if (!edds::dds::ParameterSerializer<ParameterBool_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }

    if (m_guid.guidPrefix.is_guid_static()) {
        ParameterHostIdPidInfo_t p(edds::dds::PID_HOSTID_PID_INFO, PARAMETER_HOSTID_PID_INFO_LENGTH);
        p.hostId = m_hostId;
        p.pid = m_pid;
        if (!edds::dds::ParameterSerializer<ParameterHostIdPidInfo_t>::add_to_cdr_message(p, msg)) {
            return false;
        }
    }

    return edds::dds::ParameterSerializer<Parameter_t>::add_parameter_sentinel(msg);
}

bool ParticipantProxyData::readFromCDRMessage(CDRMessage_t* msg, bool use_encapsulation,
                                              RTPSParticipantImpl* localParticipant) {
    bool are_shm_metatraffic_locators_present = false;
    bool are_shm_default_locators_present = false;
    bool is_shm_transport_possible = false;
    vbs::transport::TransportFactory* network = localParticipant->network_factory();
    bool is_shm_transport_available = localParticipant->has_shm_transport();

    auto param_process = [this, localParticipant, &network, &is_shm_transport_possible,
                          &are_shm_metatraffic_locators_present, &are_shm_default_locators_present,
                          &is_shm_transport_available](CDRMessage_t* msg_cdr, const ParameterId_t& pid,
                                                       uint16_t plength) {
        switch (pid) {
            case edds::dds::PID_KEY_HASH: {
                ParameterKey_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterKey_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                GUID_t guid;
                vbsutil::xmlparser::iHandle2GUID(guid, p.key);
                m_guid = guid;
                m_key = p.key;
                break;
            }
            case edds::dds::PID_PROTOCOL_VERSION: {
                ParameterProtocolVersion_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterProtocolVersion_t>::read_from_cdr_message(p, msg_cdr,
                                                                                                       plength)) {
                    return false;
                }

                if (p.protocolVersion.m_major < c_ProtocolVersion.m_major) {
                    return false;
                }
                m_protocolVersion = p.protocolVersion;
                break;
            }
            case edds::dds::PID_VENDORID: {
                ParameterVendorId_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterVendorId_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                m_VendorId[0] = p.vendorId[0];
                m_VendorId[1] = p.vendorId[1];
                is_shm_transport_available &= (m_VendorId == c_VendorId_vbs);
                break;
            }
            case edds::dds::PID_DOMAIN_ID: {
                ParameterDomainId_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterDomainId_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                m_domain_id = p.domain_id;
                break;
            }
            case edds::dds::PID_PARTICIPANT_GUID: {
                ParameterGuid_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterGuid_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                m_guid = p.guid;
                m_key = p.guid;
                break;
            }
            case edds::dds::PID_METATRAFFIC_MULTICAST_LOCATOR: {
                ParameterLocator_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterLocator_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                Locator_t temp_locator;
                if (network->transform_remote_locator(localParticipant->get_transport_name(), p.locator,
                                                      temp_locator)) {
                    ProxyDataFilters::filter_locators(is_shm_transport_available, &is_shm_transport_possible,
                                                      &are_shm_metatraffic_locators_present, &metatraffic_locators,
                                                      temp_locator, false);
                }
                break;
            }
            case edds::dds::PID_METATRAFFIC_UNICAST_LOCATOR: {
                ParameterLocator_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterLocator_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }
                if (p.locator.kind == LOCATOR_KIND_SHM)
                    break;
                Locator_t temp_locator;
                if (network->transform_remote_locator(localParticipant->get_transport_name(), p.locator,
                                                      temp_locator) &&
                    localParticipant->is_valid_remote_locator(p.locator)) {
                    ProxyDataFilters::filter_locators(is_shm_transport_available, &is_shm_transport_possible,
                                                      &are_shm_metatraffic_locators_present, &metatraffic_locators,
                                                      temp_locator, true);
                }

                break;
            }
            case edds::dds::PID_DEFAULT_UNICAST_LOCATOR: {
                ParameterLocator_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterLocator_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }
                if (p.locator.kind == LOCATOR_KIND_SHM)
                    break;
                if (m_VendorId != c_VendorId_vbs) {
                    default_locators.add_unicast_locator(p.locator);
                    break;
                }
                Locator_t temp_locator;
                if (network->transform_remote_locator(localParticipant->get_transport_name(), p.locator,
                                                      temp_locator) &&
                    localParticipant->is_valid_remote_locator(p.locator)) {
                    ProxyDataFilters::filter_locators(is_shm_transport_available, &is_shm_transport_possible,
                                                      &are_shm_default_locators_present, &default_locators,
                                                      temp_locator, true);
                }
                break;
            }
            case edds::dds::PID_DEFAULT_MULTICAST_LOCATOR: {
                ParameterLocator_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterLocator_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                Locator_t temp_locator;
                if (network->transform_remote_locator(localParticipant->get_transport_name(), p.locator,
                                                      temp_locator)) {
                    ProxyDataFilters::filter_locators(is_shm_transport_available, &is_shm_transport_possible,
                                                      &are_shm_default_locators_present, &default_locators,
                                                      temp_locator, false);
                }
                break;
            }
            case edds::dds::PID_PARTICIPANT_LEASE_DURATION: {
                ParameterTime_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterTime_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                m_leaseDuration = p.time.to_duration_t();
                lease_duration_ = std::chrono::microseconds(TimeConv::Duration_t2MicroSecondsInt64(m_leaseDuration));
                break;
            }
            case edds::dds::PID_BUILTIN_ENDPOINT_SET: {
                ParameterBuiltinEndpointSet_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterBuiltinEndpointSet_t>::read_from_cdr_message(p, msg_cdr,
                                                                                                          plength)) {
                    return false;
                }

                m_availableBuiltinEndpoints = p.endpointSet;
                break;
            }
            case edds::dds::PID_ENTITY_NAME: {
                ParameterString_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterString_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                m_participantName = p.getName();
                break;
            }
            case edds::dds::PID_PROPERTY_LIST: {
                if (!edds::dds::ParameterSerializer<ParameterPropertyList_t>::read_from_cdr_message(m_properties,
                                                                                                    msg_cdr, plength)) {
                    return false;
                }
                break;
            }
            case edds::dds::PID_IDENTITY_TOKEN: {
#if HAVE_SECURITY
                ParameterToken_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterToken_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                identity_token_ = std::move(p.token);
#else
                logDebug(RTPS_PARTICIPANT, "Received PID_IDENTITY_TOKEN but security is disabled");
#endif  // if HAVE_SECURITY
                break;
            }
            case edds::dds::PID_PERMISSIONS_TOKEN: {
#if HAVE_SECURITY
                ParameterToken_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterToken_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                permissions_token_ = std::move(p.token);
#else
                logDebug(RTPS_PARTICIPANT, "Received PID_PERMISSIONS_TOKEN but security is disabled");
#endif  // if HAVE_SECURITY
                break;
            }

            case edds::dds::PID_PARTICIPANT_SECURITY_INFO: {
#if HAVE_SECURITY
                ParameterParticipantSecurityInfo_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterParticipantSecurityInfo_t>::read_from_cdr_message(
                        p, msg_cdr, plength)) {
                    return false;
                }

                security_attributes_ = p.security_attributes;
                plugin_security_attributes_ = p.plugin_security_attributes;
#else
                logDebug(RTPS_PARTICIPANT, "Received PID_PARTICIPANT_SECURITY_INFO but security is disabled");
#endif  // if HAVE_SECURITY
                break;
            }
            case edds::dds::PID_EVBS_VERSION: {
                ParameterVersion_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterVersion_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                m_evbsRemoteVersion = p.evbsVersion;
                break;
            }
            case edds::dds::PID_STARTUP: {
                ParameterTime_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterTime_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                m_startTime = p.time.to_duration_t();
                break;
            }
            case edds::dds::PID_TRANSPORT_TYPE: {
                ParameterTransportType_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterTransportType_t>::read_from_cdr_message(p, msg_cdr,
                                                                                                     plength)) {
                    return false;
                }
                transport_type = (TransportTypeKind)p.type;
                break;
            }
            case edds::dds::PID_HOST_NAME: {
                ParameterString_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterString_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                m_hostName = p.getName();
                break;
            }
            case edds::dds::PID_COMPATIBLE: {
                ParameterBool_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterBool_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                isCompatible = p.value;
                break;
            }
            case edds::dds::PID_HOSTID_PID_INFO: {
                ParameterHostIdPidInfo_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterHostIdPidInfo_t>::read_from_cdr_message(p, msg_cdr,
                                                                                                     plength)) {
                    return false;
                }
                m_hostId = p.hostId;
                m_pid = p.pid;
                break;
            }
            default: {
                break;
            }
        }

        return true;
    };

    uint32_t qos_size;
    clear();
    try {
        return ParameterList::readParameterListfromCDRMsg(*msg, param_process, use_encapsulation, qos_size);
    } catch (std::bad_alloc& ba) {
        elogWarning(XTYPES, "readFromCDRMessage(Participant proxy)" << m_guid << " bad_alloc caught: " << ba.what());
        return false;
    }
}

bool ParticipantProxyData::readStartTimeFromCDRMessage(CDRMessage_t* msg, Time_t& startTime) {

    auto param_process = [this, &startTime](CDRMessage_t* msg_cdr, const ParameterId_t& pid, uint16_t plength) {
        switch (pid) {
            case edds::dds::PID_STARTUP: {
                ParameterTime_t p(pid, plength);
                if (!edds::dds::ParameterSerializer<ParameterTime_t>::read_from_cdr_message(p, msg_cdr, plength)) {
                    return false;
                }

                startTime = p.time.to_duration_t();
                break;
            }
            default: {
                break;
            }
        }

        return true;
    };

    uint32_t qos_size;
    clear();
    try {
        return ParameterList::readParameterListfromCDRMsg(*msg, param_process, true, qos_size);
    } catch (std::bad_alloc& ba) {
        elogWarning(XTYPES,
                    "readStartTimeFromCDRMessage(Participant proxy) " << m_guid << " bad_alloc caught: " << ba.what());
        return false;
    }
}

void ParticipantProxyData::clear() {
    m_protocolVersion = ProtocolVersion_t();
    m_guid = GUID_t();
    // set_VendorId_Unknown(m_VendorId);
    m_VendorId = c_VendorId_Unknown;
    m_domain_id = edds::dds::DOMAIN_ID_UNKNOWN;
    m_availableBuiltinEndpoints = 0U;
    metatraffic_locators.unicast.clear();
    metatraffic_locators.multicast.clear();
    default_locators.unicast.clear();
    default_locators.multicast.clear();
    m_participantName = "";
    m_key = InstanceHandle_t();
    m_leaseDuration = Duration_t();
    m_startTime = Time_t();
    lease_duration_ = std::chrono::microseconds::zero();
    isAlive = true;
#if HAVE_SECURITY
    identity_token_ = IdentityToken();
    permissions_token_ = PermissionsToken();
    security_attributes_ = 0UL;
    plugin_security_attributes_ = 0UL;
#endif  // if HAVE_SECURITY
    m_properties.clear();
    m_properties.length = 0U;
    m_evbsRemoteVersion = evbs::edds::rtps::c_Version_vbs_unknown;
    m_hostName = string_255();
    transport_type = TransportTypeKind::TRANSPORT_TYPE_UDP;
    isCompatible = false;
    m_hostId = 0;
    m_pid = 0;
}

void ParticipantProxyData::copy(const ParticipantProxyData& pdata) {
    m_protocolVersion = pdata.m_protocolVersion;
    m_guid = pdata.m_guid;
    m_VendorId[0] = pdata.m_VendorId[0];
    m_VendorId[1] = pdata.m_VendorId[1];
    m_availableBuiltinEndpoints = pdata.m_availableBuiltinEndpoints;
    metatraffic_locators = pdata.metatraffic_locators;
    default_locators = pdata.default_locators;
    m_participantName = pdata.m_participantName;
    m_leaseDuration = pdata.m_leaseDuration;
    lease_duration_ = std::chrono::microseconds(TimeConv::Duration_t2MicroSecondsInt64(pdata.m_leaseDuration));
    m_key = pdata.m_key;
    isAlive = pdata.isAlive;
    m_properties = pdata.m_properties;
    m_sample_identity = pdata.m_sample_identity;
    // This method is only called when a new participant is discovered.The destination of the copy
    // will always be a new ParticipantProxyData or one from the pool, so there is no need for
    // m_readers and m_writers to be copied

#if HAVE_SECURITY
    identity_token_ = pdata.identity_token_;
    permissions_token_ = pdata.permissions_token_;
    security_attributes_ = pdata.security_attributes_;
    plugin_security_attributes_ = pdata.plugin_security_attributes_;
    m_auth_token = pdata.m_auth_token;
    is_client = pdata.is_client;
#endif  // if HAVE_SECURITY
    m_startTime = pdata.m_startTime;
    m_evbsRemoteVersion = pdata.m_evbsRemoteVersion;
    isCompatible = pdata.isCompatible;
    m_hostId = pdata.m_hostId;
    m_pid = pdata.m_pid;
}

ParticipantProxyData& ParticipantProxyData::operator=(const ParticipantProxyData& pdata) {
    if (this != &pdata) {
        m_protocolVersion = pdata.m_protocolVersion;
        m_guid = pdata.m_guid;
        m_VendorId = pdata.m_VendorId;
        m_availableBuiltinEndpoints = pdata.m_availableBuiltinEndpoints;
        metatraffic_locators = pdata.metatraffic_locators;
        default_locators = pdata.default_locators;
        m_manualLivelinessCount = 0;
        m_participantName = pdata.m_participantName;
        m_key = pdata.m_key;
        m_leaseDuration = pdata.m_leaseDuration;
#if HAVE_SECURITY
        identity_token_ = pdata.identity_token_;
        permissions_token_ = pdata.permissions_token_;
        security_attributes_ = pdata.security_attributes_;
        plugin_security_attributes_ = pdata.plugin_security_attributes_;
        m_auth_token = pdata.m_auth_token;
        is_client = pdata.is_client;
#endif  // if HAVE_SECURITY
        isAlive = pdata.isAlive;
        m_properties = pdata.m_properties;
        lease_duration_event = nullptr;
        should_check_lease_duration = false;
        // This method is only called when calling the participant discovery listener and the
        // corresponding DiscoveredParticipantInfo struct is created. Only participant info is used,
        // so there is no need to copy m_readers and m_writers
        m_readers = nullptr;
        m_writers = nullptr;
        m_sample_identity = pdata.m_sample_identity;
        m_startTime = pdata.m_startTime;
        m_evbsRemoteVersion = pdata.m_evbsRemoteVersion;
        m_hostName = pdata.m_hostName;
        transport_type = pdata.transport_type;
        lease_duration_ = pdata.lease_duration_;
        isCompatible = pdata.isCompatible;
        m_hostId = pdata.m_hostId;
        m_pid = pdata.m_pid;
    }
    return *this;
}

bool ParticipantProxyData::updateData(ParticipantProxyData& pdata) {
    metatraffic_locators = pdata.metatraffic_locators;
    default_locators = pdata.default_locators;
    m_leaseDuration = pdata.m_leaseDuration;
    isAlive = true;
    m_properties = pdata.m_properties;
#if HAVE_SECURITY
    identity_token_ = pdata.identity_token_;
    permissions_token_ = pdata.permissions_token_;
    security_attributes_ = pdata.security_attributes_;
    plugin_security_attributes_ = pdata.plugin_security_attributes_;
    m_auth_token = pdata.m_auth_token;
    is_client = pdata.is_client;
#endif  // if HAVE_SECURITY
    m_startTime = pdata.m_startTime;
    m_evbsRemoteVersion = pdata.m_evbsRemoteVersion;
    isCompatible = pdata.isCompatible;
    auto new_lease_duration = std::chrono::microseconds(TimeConv::Duration_t2MicroSecondsInt64(m_leaseDuration));
    if (lease_duration_event != nullptr) {
        if (new_lease_duration < lease_duration_) {
            // Calculate next trigger.
            auto real_lease_tm = last_received_message_tm_ + new_lease_duration;
            auto next_trigger = real_lease_tm - std::chrono::steady_clock::now();
            lease_duration_event->cancel_timer();
            (void)lease_duration_event->update_interval_millisec(
                (double_t)std::chrono::duration_cast<std::chrono::milliseconds>(next_trigger).count());
            lease_duration_event->restart_timer();
        }
    }
    lease_duration_ = new_lease_duration;
    m_hostId = pdata.m_hostId;
    m_pid = pdata.m_pid;
    return true;
}

void ParticipantProxyData::set_persistence_guid(const GUID_t& guid) {
    // only valid values
    if (guid == c_Guid_Unknown) {
        return;
    }

    // generate pair
    std::pair<std::string, std::string> persistent_guid;
    persistent_guid.first = edds::dds::parameter_property_persistence_guid;

    std::ostringstream data;
    data << guid;
    persistent_guid.second = data.str();

    // if exists replace
    ParameterPropertyList_t::iterator it = std::find_if(
        m_properties.begin(), m_properties.end(),
        [&persistent_guid](const edds::dds::ParameterProperty_t& p) { return persistent_guid.first == p.first(); });

    if (it != m_properties.end()) {
        if (!it->modify(persistent_guid)) {
            elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ERROR,
                      "Failed to change property <" << it->first() << " | " << it->second() << "> to <"
                                                    << persistent_guid.first << " | " << persistent_guid.second << ">");
        }
    } else {
        // if not exists add
        (void)m_properties.push_back(persistent_guid.first, persistent_guid.second);
    }
}

GUID_t ParticipantProxyData::get_persistence_guid() const {
    GUID_t persistent(c_Guid_Unknown);

    ParameterPropertyList_t::const_iterator it =
        std::find_if(m_properties.begin(), m_properties.end(), [](const edds::dds::ParameterProperty_t p) {
            return edds::dds::parameter_property_persistence_guid == p.first();
        });

    if (it != m_properties.end()) {
        std::istringstream in(it->second());
        in >> persistent;
    }

    return persistent;
}

std::string ParticipantProxyData::get_participant_type() const {
    ParameterPropertyList_t::const_iterator participant_type =
        std::find_if(m_properties.begin(), m_properties.end(), [](const edds::dds::ParameterProperty_t p) {
            return (edds::dds::parameter_property_participant_type == p.first() ||
                    edds::dds::parameter_property_participant_type_1_0 == p.first());
        });

    if (participant_type != m_properties.end()) {
        if (participant_type->second() == edds::rtps::ParticipantType::SERVER) {
            return edds::rtps::ParticipantType::SERVER;
        } else if (participant_type->second() == edds::rtps::ParticipantType::BACKUP) {
            return edds::rtps::ParticipantType::BACKUP;
        } else if (participant_type->second() == edds::rtps::ParticipantType::SIMPLE) {
            return edds::rtps::ParticipantType::SIMPLE;
        } else if (participant_type->second() == edds::rtps::ParticipantType::CLIENT) {
            return edds::rtps::ParticipantType::CLIENT;
        } else if (participant_type->second() == edds::rtps::ParticipantType::SUPER_CLIENT) {
            return edds::rtps::ParticipantType::SUPER_CLIENT;
        } else {
            elogError(RTPS_PARTICIPANT, RetCode_t::RETCODE_ERROR,
                      "Wrong " << edds::dds::parameter_property_participant_type << ": " << participant_type->second());
            return edds::rtps::ParticipantType::UNKNOWN;
        }
    } else {
        logDebug(RTPS_PARTICIPANT, edds::dds::parameter_property_participant_type << " is not set");
        return edds::rtps::ParticipantType::SIMPLE;
    }
}

void ParticipantProxyData::set_sample_identity(const SampleIdentity& sid) {
    edds::dds::set_proxy_property(sid, "PID_CLIENT_SERVER_KEY", m_properties);
}

SampleIdentity ParticipantProxyData::get_sample_identity() const {
    return edds::dds::get_proxy_property<SampleIdentity>("PID_CLIENT_SERVER_KEY", m_properties);
}

void ParticipantProxyData::set_backup_stamp(const GUID_t& guid) {
    edds::dds::set_proxy_property(guid, "PID_BACKUP_STAMP", m_properties);
}

GUID_t ParticipantProxyData::get_backup_stamp() const {
    return edds::dds::get_proxy_property<GUID_t>("PID_BACKUP_STAMP", m_properties);
}

void ParticipantProxyData::assert_liveliness() {
    last_received_message_tm_ = std::chrono::steady_clock::now();
}

EvbsVersion_t ParticipantProxyData::get_evbs_remote_version() {
    return m_evbsRemoteVersion;
}

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */
